package com.demo.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.demo.rocketmq.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;

/**
 * rocketmq 生产者，主要是展示生产者发消息的功能
 * @author chenyongjiu
 */
@Slf4j
@Component
public class Producer {

    @Resource(name = "rocketMQTemplate")
    private RocketMQTemplate rocketMqTemplate;

    public void sendMsg() {
        // 这种方式具有内部重试机制，即在主动声明本次消息发送失败之前，内部实现将重试一定次数，默认为2次，向不同broker重试
        // 发送的结果存在同一个消息可能被多次发送给给broker，这里需要应用的开发者自己在消费端处理幂等性问题。
        // convertAndSend 和 send 方法底层都是调用 syncSend 方法
        // 默认超时时间是3000毫秒
        Date date = new Date();
        log.info("发送日期：" + date);
        rocketMqTemplate.convertAndSend("test-topic-1", "Hello, World! 日期：" + date);
        rocketMqTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message! 日期:" + date).build());
    }

    public void syncSend() {
        // timeout不能设置太短，否则容易出现异常
        SendResult sendResult = rocketMqTemplate.syncSend("test-topic-1", "Hello, World! I'm from simple message", 1000L);
        log.info("syncSend 结果：{}", JSON.toJSONString(sendResult));
    }

    /**
     * 使用topic：ORDER，test-topic-2
     */
    public void convertAndSend() {
        // topic: ORDER，tag: paid, cancel
        // 同一主题下区分不同类型的消息
        rocketMqTemplate.convertAndSend("ORDER:paid", "Hello, World! paid");
        rocketMqTemplate.convertAndSend("ORDER:cancel", "Hello, World! cancel");
        // 消息体为自定义对象
        rocketMqTemplate.convertAndSend("test-topic-2", new User("name"));
    }

    public void sendAndReceive() {
        // 同步发送request并且等待user类型的返回值
        Object object = rocketMqTemplate.sendAndReceive("test-topic-3", "I'm sendAndReceive message", User.class);
        // 打印消费端返回的结果
        log.info("Is MessageExt Object:{}, object:{}", object instanceof User, JSON.toJSONString(object));
    }

    public void sendDelay() {
        // delayLevel是延迟等级，延迟等级对应是配置文件里的messageDelayLevel，
        // messageDelayLevel默认值是1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        rocketMqTemplate.syncSend("test-topic-1", MessageBuilder.withPayload("I'm delayed message").build(), 10000, 3);
    }

    public void sendOneWay() {
        // 消息发送给broker立即返回，不用等待来自broker的ack来告知本次消息发送是否完全完成发送。
        // 这种方式吞吐量很大，但是存在消息丢失的风险，所以其适用于不重要的消息发送，比如日志收集
        rocketMqTemplate.sendOneWay("test-topic-1", MessageBuilder.withPayload("I'm one way message").build());
        rocketMqTemplate.sendOneWay("test-topic-1", "I'm one way message part2");
    }

    public void syncSendOrderly() {
        // 发送顺序消息，通过hashKey将消息发送到固定的队列，保证消息是顺序
        rocketMqTemplate.syncSendOrderly("test-topic-1", "I'm order message", "1234");
    }

    public void asyncSend() {
        // 发送消息采用异步发送模式，消息发送后立刻返回，当消息完全完成发送后，会调用回调函数sendCallback
        // 来告知发送者本次发送成功或者失败。异步模式通常用于响应时间敏感业务场景，即承受不了同步发送消息时等待返回的耗时代价。
        // 同同步发送一样，异步模式也在内部实现了重试机制，默认次数为2次，是同一个broker重试
        rocketMqTemplate.asyncSend("test-topic-1", MessageBuilder.withPayload("I'm one async message").build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送异步消息成功: {}", JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("发送异步消息失败！！", e);
            }
        });
    }

    /**
     * 发送RocketMQ事务消息
     * 自定义事务消息实现类，必须实现RocketMQLocalTransactionListener接口，而且指定rocketTemplate
     */
    public TransactionSendResult transactionMsgWithTx(Object param) {
        User user = new User();
        user.setUserName("withTx");
        user.setId(2L);

        Message<User> message = MessageBuilder.withPayload(user).build();
        return rocketMqTemplate.sendMessageInTransaction("test-topic-2", message, param);
    }

    /**
     * 测试消费端消费失败，broker重试16次，然后将消息放到死信队列
     */
    public void sendConsumerFailMsg() {
        rocketMqTemplate.convertAndSend("consumer-fail-topic", "fail msg, 时间：" + new Date());
    }
}
